package com.saic.utils;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.security.UserGroupInformation;

import com.google.gson.Gson;

public class HbaseUtil {
	
	public static int count  = 0;
	
	public static List<Put> putList  = new ArrayList<Put>();
	public static long time1;
	private static String family;
	private static InputStream in = null;
	public static Configuration conf = null;
	public static Connection connection = null;
	public static Admin admin = null;
	public static Map<String, Table> tableMap = new HashMap<String, Table>();
	private static String hbaseKrbUser = null;
	private static String hmasterPrincipal = null;
	private static String hregionserverPrincipal = null;
	private static String krbFile = null;
	private static String keytabFile = null;
	private static String hmaster = null;
	private static String hbase_zookeeper_quorum = null;
	private static String clientPort = null;
	// initialize variable
	static {
		try {
			in = HbaseUtil.class.getClassLoader().getResourceAsStream("hbaseconf.properties");
			Properties properties = new Properties();
			properties.load(in);
			family = properties.getProperty("familyName");
			hmaster = properties.getProperty("hbase_master");
			hbase_zookeeper_quorum = properties.getProperty("hbase_zookeeper_quorum");
			clientPort = properties.getProperty("hbase_zookeeper_property_clientPort");
			
			String krbEnable = properties.getProperty("use_kerberos_authentication", "false").toLowerCase();
			if(krbEnable.equals("true")){
				hmasterPrincipal = properties.getProperty("hbase_master_kerberos_principal");
				hregionserverPrincipal = properties.getProperty("hbase_regionserver_kerberos_principal");
				krbFile = properties.getProperty("krbFile");
				keytabFile = properties.getProperty("keytab_path");
				hbaseKrbUser = properties.getProperty("hbaseKrbUser");
				
				System. setProperty("java.security.krb5.conf", krbFile);
			}
			//
			conf = HBaseConfiguration.create();
			conf.set("hbase.zookeeper.quorum", hbase_zookeeper_quorum);
			conf.set("hbase.zookeeper.property.clientPort",clientPort);
			conf.set("hbase.master", hmaster);
			
			if(krbEnable.equals("true")){
				//Kerberos authentication
				conf.set("hadoop.security.authentication" , "Kerberos" );
				conf.set("hbase.security.authentication" , "Kerberos" );
				conf.set("hbase.zookeeper.client.keytab.file" ,keytabFile);
			       
				conf.set("hbase.master.kerberos.principal",hmasterPrincipal);
				conf.set("hbase.master.keytab.file" ,keytabFile);
				conf.set("hbase.regionserver.kerberos.principal",hregionserverPrincipal);
				conf.set("hbase.regionserver.keytab.file" ,keytabFile);
				
				UserGroupInformation.setConfiguration(conf);
			    UserGroupInformation.loginUserFromKeytab(hbaseKrbUser,keytabFile);
				}
			
			connection = ConnectionFactory.createConnection(conf);
			admin = connection.getAdmin();
		} catch (Exception e) {

		} finally {
			try {
				in.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	

	/**
	 * 
	 * @param JSONString
	 * @return
	 */
	 public static Map<String,Object> JSONStringToMap(String JSONString){
	        Map<String,Object> map  = new HashMap<String, Object>();
	        Gson gson = new Gson();
	        java.lang.reflect.Type type = new com.google.gson.reflect.TypeToken<Map<String, Object>>(){}.getType();
	        map = gson.fromJson(JSONString, type);
	        for(String key:map.keySet()){
	        	System.err.println("key: "+key+"   value: "+map.get(key));
	        }
	        return map;

	 }
	 /**
	  * 按指定方式切分字符串构造RowKey
	  * @param map
	  * @param string
	  * @return
	  */
	 public static String CreateRowKey (Map<String,String> map,String rowStr){
		StringBuffer rowKey = new StringBuffer();
		String[] split = rowStr.split("\\|");
		int count = 0;
		for(String s:split){
			rowKey.append(map.get(s));
			if((++count) <split.length ){
			rowKey.append("||");	
			}
		}
		return rowKey.toString();
		 
	 }
	
	 public static void putData(Map<String,String> map){
		 String tableName = map.get("tableName");
		 Table table = getTable(tableName);
		 String rowStr = map.get("rowkey");
		 String rowKey = CreateRowKey(map, rowStr);
		 Put put = new Put(rowKey.getBytes());
		 map.remove("rowkey");
		 map.remove("tableName");
		 for(String key:map.keySet()){
			 if("null".equals(map.get(key))){
		        put.addColumn(family.getBytes(),key.getBytes(), "NULL".getBytes());		       			       		   
			 }else{
	           put.addColumn(family.getBytes(),key.getBytes(), map.get(key).toString().getBytes());		       			       		   
			 }
		 }
		try {
			table.put(put);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			System.out.println("put sucess");
		}
	 }
	public  static void putList(Table table,Put put){
		long time2 = System.currentTimeMillis();
		putList.add(put);
		if(putList.size() == 100 || (time2-time1)>1000*10){
			try {
				System.out.println("putLis： "+putList.size());
				table.put(putList);
			} catch (IOException e) {
				e.printStackTrace();
			}finally {
				putList.clear();
				time1 = System.currentTimeMillis();
			}
		}
		
	}
	/**
	 * 
	 * @param tableName
	 * @return
	 */
	public   static Table getTable(String tableName) {
		Table table = tableMap.get(tableName);
		if (null == table) {
			table = createTable(tableName);
		}
		return table;
	}
    public  synchronized  static Table createTable(String tableName){
    	Table table = tableMap.get(tableName);
    	try {
			TableName tableNameObj = TableName.valueOf(tableName);
			if (!admin.tableExists(tableNameObj)){
				HTableDescriptor tableDescriptor = new HTableDescriptor(tableNameObj);
				tableDescriptor.addFamily(new HColumnDescriptor(family));
				admin.createTable(tableDescriptor);
				table = connection.getTable(TableName.valueOf(tableName));
				tableMap.put(tableName, table);
			}
			else{
			table = connection.getTable(TableName.valueOf(tableName));
			tableMap.put(tableName, table);
			}
		
		} catch (IOException e) {
			e.printStackTrace();
		}
    	return table;
    }
	
}
